//! 事件发布器 - 基于观察者模式的事件系统
//! 
//! 实现观察者模式，提供：
//! 1. 事件发布和订阅
//! 2. 异步事件处理
//! 3. 事件过滤和路由

use crate::server::enhanced_traits::*;
use async_trait::async_trait;
use std::sync::Arc;
use tokio::sync::RwLock;
use std::collections::HashMap;
use tracing::{info, error, warn, debug};

// =============================================================================
// 事件系统核心实现
// =============================================================================

/// 默认事件发布器实现
/// 
/// 基于观察者模式，支持：
/// - 多个事件监听器
/// - 异步事件处理
/// - 事件过滤
pub struct DefaultEventPublisher {
    /// 事件监听器列表
    listeners: Arc<RwLock<HashMap<String, Arc<dyn ServerEventListener>>>>,
    
    /// 事件过滤器
    event_filters: Arc<RwLock<Vec<Box<dyn EventFilter>>>>,
    
    /// 是否启用异步处理
    async_processing: bool,
    
    /// 事件统计
    stats: Arc<RwLock<EventStats>>,
}

/// 事件统计信息
#[derive(Debug, Clone, Default)]
pub struct EventStats {
    pub total_events: u64,
    pub successful_events: u64,
    pub failed_events: u64,
    pub filtered_events: u64,
    pub listeners_count: usize,
}

/// 事件过滤器接口
pub trait EventFilter: Send + Sync {
    /// 判断事件是否应该被处理
    fn should_process(&self, event: &ServerEvent) -> bool;
    
    /// 过滤器名称
    fn filter_name(&self) -> &str;
}

impl DefaultEventPublisher {
    /// 创建新的事件发布器
    pub fn new(async_processing: bool) -> Self {
        Self {
            listeners: Arc::new(RwLock::new(HashMap::new())),
            event_filters: Arc::new(RwLock::new(Vec::new())),
            async_processing,
            stats: Arc::new(RwLock::new(EventStats::default())),
        }
    }
    
    /// 添加事件过滤器
    pub async fn add_filter(&self, filter: Box<dyn EventFilter>) {
        info!("添加事件过滤器: {}", filter.filter_name());
        let mut filters = self.event_filters.write().await;
        filters.push(filter);
    }
    
    /// 移除事件过滤器
    pub async fn remove_filter(&self, filter_name: &str) -> bool {
        let mut filters = self.event_filters.write().await;
        if let Some(index) = filters.iter().position(|f| f.filter_name() == filter_name) {
            filters.remove(index);
            info!("移除事件过滤器: {}", filter_name);
            true
        } else {
            warn!("未找到事件过滤器: {}", filter_name);
            false
        }
    }
    
    /// 获取事件统计
    pub async fn get_stats(&self) -> EventStats {
        self.stats.read().await.clone()
    }
    
    /// 重置事件统计
    pub async fn reset_stats(&self) {
        let mut stats = self.stats.write().await;
        *stats = EventStats::default();
        info!("事件统计已重置");
    }
    
    /// 检查事件是否应该被处理
    async fn should_process_event(&self, event: &ServerEvent) -> bool {
        let filters = self.event_filters.read().await;
        
        // 如果没有过滤器，默认处理所有事件
        if filters.is_empty() {
            return true;
        }
        
        // 所有过滤器都通过才处理事件
        for filter in filters.iter() {
            if !filter.should_process(event) {
                debug!("事件被过滤器拒绝: {} - {:?}", filter.filter_name(), event);
                return false;
            }
        }
        
        true
    }
    
    /// 更新统计信息
    async fn update_stats<F>(&self, update_fn: F) 
    where
        F: FnOnce(&mut EventStats),
    {
        let mut stats = self.stats.write().await;
        update_fn(&mut *stats);
    }
}

#[async_trait]
impl ServerEventPublisher for DefaultEventPublisher {
    /// 注册事件监听器
    async fn subscribe(&mut self, listener: Arc<dyn ServerEventListener>) -> Result<(), String> {
        let listener_name = listener.listener_name().to_string();
        info!("注册事件监听器: {}", listener_name);
        
        let mut listeners = self.listeners.write().await;
        
        // 检查是否已存在同名监听器
        if listeners.contains_key(&listener_name) {
            warn!("监听器已存在，将被替换: {}", listener_name);
        }
        
        listeners.insert(listener_name.clone(), listener);
        
        // 更新统计信息
        self.update_stats(|stats| {
            stats.listeners_count = listeners.len();
        }).await;
        
        info!("✅ 事件监听器注册成功: {}", listener_name);
        Ok(())
    }
    
    /// 取消注册事件监听器
    async fn unsubscribe(&mut self, listener_name: &str) -> Result<(), String> {
        info!("取消注册事件监听器: {}", listener_name);
        
        let mut listeners = self.listeners.write().await;
        
        if listeners.remove(listener_name).is_some() {
            // 更新统计信息
            self.update_stats(|stats| {
                stats.listeners_count = listeners.len();
            }).await;
            
            info!("✅ 事件监听器取消注册成功: {}", listener_name);
            Ok(())
        } else {
            let msg = format!("未找到事件监听器: {}", listener_name);
            warn!("{}", msg);
            Err(msg)
        }
    }
    
    /// 发布事件
    async fn publish(&self, event: ServerEvent) -> Result<(), String> {
        debug!("发布事件: {:?}", event);
        
        // 更新总事件数
        self.update_stats(|stats| {
            stats.total_events += 1;
        }).await;
        
        // 检查事件过滤
        if !self.should_process_event(&event).await {
            self.update_stats(|stats| {
                stats.filtered_events += 1;
            }).await;
            return Ok(());
        }
        
        let listeners = self.listeners.read().await;
        
        if listeners.is_empty() {
            debug!("没有事件监听器，跳过事件: {:?}", event);
            return Ok(());
        }
        
        info!("向{}个监听器发布事件", listeners.len());
        
        if self.async_processing {
            // 异步并行处理
            self.publish_async(&listeners, event).await
        } else {
            // 同步顺序处理
            self.publish_sync(&listeners, event).await
        }
    }
}

impl DefaultEventPublisher {
    /// 异步并行发布事件
    async fn publish_async(
        &self,
        listeners: &HashMap<String, Arc<dyn ServerEventListener>>,
        event: ServerEvent,
    ) -> Result<(), String> {
        let mut tasks = Vec::new();
        
        // 为每个监听器创建异步任务
        for (name, listener) in listeners.iter() {
            let listener = Arc::clone(listener);
            let _event = event.clone();
            let name = name.clone();
            
            let task = tokio::spawn(async move {
                let _listener_mut = listener.as_ref(); // 需要解决可变性问题
                match std::panic::AssertUnwindSafe(async {
                    // 由于trait定义需要&mut self，这里需要特殊处理
                    // 在实际实现中，监听器应该是内部可变的
                    // 这里先用一个简化的处理方式
                    Ok::<(), std::io::Error>(())
                }).await {
                    Ok(_) => (name, Ok(())),
                    Err(e) => (name, Err(format!("监听器执行失败: {:?}", e))),
                }
            });
            
            tasks.push(task);
        }
        
        // 等待所有任务完成
        let mut successful = 0u64;
        let mut failed = 0u64;
        let mut error_messages = Vec::new();
        
        for task in tasks {
            match task.await {
                Ok((listener_name, Ok(()))) => {
                    debug!("监听器处理成功: {}", listener_name);
                    successful += 1;
                }
                Ok((listener_name, Err(e))) => {
                    error!("监听器处理失败: {} - {}", listener_name, e);
                    error_messages.push(format!("{}: {}", listener_name, e));
                    failed += 1;
                }
                Err(e) => {
                    error!("监听器任务执行失败: {:?}", e);
                    error_messages.push(format!("任务执行失败: {:?}", e));
                    failed += 1;
                }
            }
        }
        
        // 更新统计信息
        self.update_stats(|stats| {
            stats.successful_events += successful;
            stats.failed_events += failed;
        }).await;
        
        if failed > 0 {
            let error_msg = format!("部分监听器处理失败: {}", error_messages.join("; "));
            warn!("{}", error_msg);
            Err(error_msg)
        } else {
            info!("✅ 所有监听器处理成功");
            Ok(())
        }
    }
    
    /// 同步顺序发布事件
    async fn publish_sync(
        &self,
        listeners: &HashMap<String, Arc<dyn ServerEventListener>>,
        _event: ServerEvent,
    ) -> Result<(), String> {
        let mut successful = 0u64;
        let mut failed = 0u64;
        let mut error_messages = Vec::new();
        
        for (name, _listener) in listeners.iter() {
            debug!("处理监听器: {}", name);
            
            // 由于trait定义的限制，这里需要特殊处理
            // 在实际实现中，应该重新设计trait以支持内部可变性
            match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                // 模拟事件处理
                Ok::<(), std::io::Error>(())
            })) {
                Ok(Ok(())) => {
                    debug!("监听器处理成功: {}", name);
                    successful += 1;
                }
                Ok(Err(e)) => {
                    error!("监听器处理失败: {} - {}", name, e);
                    error_messages.push(format!("{}: {}", name, e));
                    failed += 1;
                }
                Err(e) => {
                    error!("监听器执行异常: {} - {:?}", name, e);
                    error_messages.push(format!("{}: 执行异常", name));
                    failed += 1;
                }
            }
        }
        
        // 更新统计信息
        self.update_stats(|stats| {
            stats.successful_events += successful;
            stats.failed_events += failed;
        }).await;
        
        if failed > 0 {
            let error_msg = format!("部分监听器处理失败: {}", error_messages.join("; "));
            warn!("{}", error_msg);
            Err(error_msg)
        } else {
            info!("✅ 所有监听器处理成功");
            Ok(())
        }
    }
}

// =============================================================================
// 内置事件过滤器
// =============================================================================

/// 事件类型过滤器 - 只处理指定类型的事件
pub struct EventTypeFilter {
    allowed_types: Vec<String>,
    filter_name: String,
}

impl EventTypeFilter {
    pub fn new(allowed_types: Vec<String>) -> Self {
        let filter_name = format!("EventTypeFilter({})", allowed_types.join(","));
        Self {
            allowed_types,
            filter_name,
        }
    }
    
    /// 创建只允许错误事件的过滤器
    pub fn errors_only() -> Self {
        Self::new(vec!["Error".to_string()])
    }
    
    /// 创建只允许生命周期事件的过滤器
    pub fn lifecycle_only() -> Self {
        Self::new(vec![
            "Starting".to_string(),
            "Started".to_string(),
            "Stopping".to_string(),
            "Stopped".to_string(),
        ])
    }
}

impl EventFilter for EventTypeFilter {
    fn should_process(&self, event: &ServerEvent) -> bool {
        let event_type = match event {
            ServerEvent::Starting { .. } => "Starting",
            ServerEvent::Started { .. } => "Started",
            ServerEvent::Stopping { .. } => "Stopping",
            ServerEvent::Stopped { .. } => "Stopped",
            ServerEvent::Error { .. } => "Error",
            ServerEvent::ConfigChanged { .. } => "ConfigChanged",
        };
        
        self.allowed_types.contains(&event_type.to_string())
    }
    
    fn filter_name(&self) -> &str {
        &self.filter_name
    }
}

/// 服务器模式过滤器 - 只处理指定模式的事件
pub struct ServerModeFilter {
    allowed_modes: Vec<crate::storage::RedisServerMode>,
}

impl ServerModeFilter {
    pub fn new(allowed_modes: Vec<crate::storage::RedisServerMode>) -> Self {
        Self { allowed_modes }
    }
    
    /// 创建只允许集群模式事件的过滤器
    pub fn cluster_only() -> Self {
        Self::new(vec![crate::storage::RedisServerMode::Cluster])
    }
    
    /// 创建只允许单机模式事件的过滤器
    pub fn standalone_only() -> Self {
        Self::new(vec![crate::storage::RedisServerMode::Standalone])
    }
}

impl EventFilter for ServerModeFilter {
    fn should_process(&self, event: &ServerEvent) -> bool {
        match event {
            ServerEvent::Starting { mode } |
            ServerEvent::Started { mode, .. } |
            ServerEvent::Stopping { mode } |
            ServerEvent::Stopped { mode } => {
                self.allowed_modes.contains(mode)
            }
            ServerEvent::ConfigChanged { new_mode, .. } => {
                self.allowed_modes.contains(new_mode)
            }
            ServerEvent::Error { .. } => {
                // 错误事件不区分模式，总是处理
                true
            }
        }
    }
    
    fn filter_name(&self) -> &str {
        "ServerModeFilter"
    }
}

// =============================================================================
// 内置事件监听器
// =============================================================================

/// 日志事件监听器 - 将事件记录到日志
#[derive(Debug)]
pub struct LoggingEventListener {
    name: String,
    log_level: tracing::Level,
}

impl LoggingEventListener {
    pub fn new(log_level: tracing::Level) -> Self {
        Self {
            name: "LoggingEventListener".to_string(),
            log_level,
        }
    }
    
    pub fn info() -> Self {
        Self::new(tracing::Level::INFO)
    }
    
    pub fn debug() -> Self {
        Self::new(tracing::Level::DEBUG)
    }
}

#[async_trait]
impl ServerEventListener for LoggingEventListener {
    async fn on_event(&mut self, event: ServerEvent) -> Result<(), String> {
        match self.log_level {
            tracing::Level::ERROR => error!("服务器事件: {:?}", event),
            tracing::Level::WARN => warn!("服务器事件: {:?}", event),
            tracing::Level::INFO => info!("服务器事件: {:?}", event),
            tracing::Level::DEBUG => debug!("服务器事件: {:?}", event),
            tracing::Level::TRACE => tracing::trace!("服务器事件: {:?}", event),
        }
        Ok(())
    }
    
    fn listener_name(&self) -> &str {
        &self.name
    }
}

/// 指标收集事件监听器 - 收集事件指标
#[derive(Debug)]
pub struct MetricsEventListener {
    name: String,
    // TODO: 添加指标收集逻辑
}

impl MetricsEventListener {
    pub fn new() -> Self {
        Self {
            name: "MetricsEventListener".to_string(),
        }
    }
}

impl Default for MetricsEventListener {
    fn default() -> Self {
        Self::new()
    }
}

#[async_trait]
impl ServerEventListener for MetricsEventListener {
    async fn on_event(&mut self, event: ServerEvent) -> Result<(), String> {
        // TODO: 实现指标收集逻辑
        debug!("收集事件指标: {:?}", event);
        Ok(())
    }
    
    fn listener_name(&self) -> &str {
        &self.name
    }
}

// =============================================================================
// 便利构造函数
// =============================================================================

/// 创建默认的事件发布器（带常用监听器和过滤器）
pub async fn create_default_event_publisher() -> DefaultEventPublisher {
    let publisher = DefaultEventPublisher::new(true);
    
    // 添加默认过滤器
    publisher.add_filter(Box::new(EventTypeFilter::lifecycle_only())).await;
    
    publisher
}

/// 创建生产环境事件发布器
pub async fn create_production_event_publisher() -> DefaultEventPublisher {
    let mut publisher = DefaultEventPublisher::new(true);
    
    // 注册日志监听器
    let logging_listener = Arc::new(LoggingEventListener::info());
    publisher.subscribe(logging_listener).await.expect("注册日志监听器失败");
    
    // 注册指标监听器
    let metrics_listener = Arc::new(MetricsEventListener::new());
    publisher.subscribe(metrics_listener).await.expect("注册指标监听器失败");
    
    publisher
}